package com.cmsz.collection.service;

import com.cmsz.collection.bean.AlarmLog;
import com.cmsz.collection.util.DateUtil;
import net.sf.json.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.List;

/**
 * @author liupeng
 * @date 2019/12/12
 * @description 自动监听消费kafka
 */

@Component
public class KafkaListenerAlarmService {

    private static Logger logger = LoggerFactory.getLogger(KafkaListenerAlarmService.class);

    @Autowired
    AlarmLogService alarmLogService;

    /** 这里是批量消费alarm入mysql */
    @KafkaListener(topics = "${spring.kafka.listener.topic}",
            containerFactory = "kafkaListenerContainerFactory2",
            groupId = "${spring.kafka.listener.mysql.group-id}")
    public void listenAlarm(List<ConsumerRecord> records, Acknowledgment ack){
        logger.info("进入alarm入库mysql");
        List<AlarmLog> list = new ArrayList<>();
        for (ConsumerRecord record : records) {
            JSONObject value = JSONObject.fromObject(record.value());
            String content = (String) value.get("message");
            JSONObject fields = JSONObject.fromObject(value.get("fields"));
            String appName = (String) fields.get("applicationName");
            list.add(new AlarmLog(appName,content, DateUtil.getTime()));
        }
        ack.acknowledge();
    }



}
